package rpc_protobuf

import (
	"context"
	"goland-study/go_code/concurrency_design_model/pubsub/impl1"
	"io"
	"strings"
	"time"
)

type HelloServiceImpl struct {
}

func (s *HelloServiceImpl) Hello(ctx context.Context, req *String) (*String, error) {
	return &String{Value: "server Hello response: " + req.Value}, nil
}

func (s *HelloServiceImpl) Channel(stream HelloService_ChannelServer) error {
	for {
		msg, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				return nil
			}
			return err
		}

		reply := &String{Value: "server Channel response:" + msg.GetValue()}

		err = stream.Send(reply)
		if err != nil {
			return err
		}
	}
}

func (s *HelloServiceImpl) mustEmbedUnimplementedHelloServiceServer() {}

type PubsubServerImpl struct {
	pub *impl1.Publisher
}

func NewPubsubServerImpl() *PubsubServerImpl {
	return &PubsubServerImpl{
		pub: impl1.NewPublisher(100*time.Millisecond, 10),
	}
}

func (p *PubsubServerImpl) Publish(ctx context.Context, msg *String) (*String, error) {
	p.pub.Publish(msg.GetValue())
	return &String{}, nil
}

func (p *PubsubServerImpl) Subscribe(msg *String, stream PubsubService_SubscribeServer) error {
	ch := p.pub.SubscribeTopic(func(v interface{}) bool {
		if v, ok := v.(string); ok {
			if strings.HasPrefix(v, msg.GetValue()) {
				return true
			}
		}
		return false
	})

	for i := range ch {
		if err := stream.Send(&String{Value: i.(string)}); err != nil {
			return err
		}
	}

	return nil
}

func (p *PubsubServerImpl) mustEmbedUnimplementedPubsubServiceServer() {}
